package cn.xuexiyuan.flinkstudy.hello;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-1-22
 * @Version 1.0
 **/
public class WordCount {

    public static void main(String[] args) throws Exception {
        // 创建 Flink 执行环境的上下文
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<String> lines = env.fromElements(
                "JAVA C# Python C++ Flink",
                "Java hello world.",
                "python hello world.",
                "Flink hello world."
        );

        DataSet<Tuple2<String, Integer>> counts = lines.flatMap(new LineSplitter())
                .groupBy(0)
                .sum(1);

        counts.print();
    }

    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String[] tokens = s.toLowerCase().split("\\W+");
            for (String token : tokens) {
                if (token.length() > 0) {
                    collector.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}
